package org.lds.simple.disruptor;

import lombok.Data;
import org.lds.simple.disruptor.event.Event;
import org.lds.simple.disruptor.event.EventFactory;
import org.lds.simple.disruptor.producer.sequencer.Sequencer;

@Data
public class RingBuffer<T extends Event> implements LifeCycle {
    private final Object[] events;

    private final int bufferSize;
    private final int bufferMask;
    private final EventFactory eventFactory;

    /**
     * 生产者
     */
    private final Sequencer sequencer;

    private boolean shutdown = false;

    public RingBuffer(int bufferSize, EventFactory eventFactory, Sequencer sequencer) {
        this.bufferSize = buildBufferSize(bufferSize);
        this.bufferMask = bufferSize - 1;
        this.eventFactory = eventFactory;
        this.events = buildEvents(bufferSize);
        this.sequencer = sequencer;
    }

    private Object[] buildEvents(int bufferSize) {
        int eventsSize = buildBufferSize(bufferSize);
        Object[] events = new Object[eventsSize];
        for (int i = 0; i < events.length; i++) {
            events[i] = eventFactory.newInstance();
        }
        return events;
    }

    /**
     * 2的幂次方
     * @param bufferSize
     */
    private int buildBufferSize(int bufferSize) {
        if (bufferSize < 2 || bufferSize > 1024 * 1024) return 1024;
        int i = bufferSize & (bufferSize - 1);
        if (i == 0) return bufferSize;
        return 1024;
    }

    public <T> T getEvent(long sequence) {
        return (T) events[index(sequence)];
    }

    private int index(long sequence) {
        return (int)sequence & bufferMask;
    }

    public long next() {
        if (this.shutdown) return Long.MIN_VALUE;
        return sequencer.next();
    }

    @Override
    public void start() {

    }

    @Override
    public void shutdown() {
        this.shutdown = true;
    }

    public void publish(long producerSequence) {
        sequencer.updateSequence(producerSequence);
        sequencer.notifyConsumer();
    }

}
